package org.yonggan.shop.rdd.report

import com.google.gson.Gson
import org.apache.commons.lang.StringUtils
import org.slf4j.LoggerFactory
import org.yonggan.shop.constant.ConfigurationManager
import org.yonggan.shop.domain.LogJson
import org.yonggan.shop.utils.SparkUtils

import scala.collection.mutable

/**
  * 各个分类（手机、书）商品页面被浏览次数最多的TON2
  */
object RptGroupedTopN {

  private val LOGGER = LoggerFactory.getLogger(RptGroupedTopN.getClass)

  def main(args: Array[String]): Unit = {

    val spark = SparkUtils.getSparkSession("商品页面被浏览次数最多的TON2")
    val sc = spark.sparkContext
    // 文件读取
    val baseRDD = sc.textFile(ConfigurationManager.TASK_INJSON)
      .map(line => {
        var logJson: LogJson = null
        val gson = new Gson()
        try {
          logJson = gson.fromJson(line, classOf[LogJson])
        } catch {
          case e: Exception => {
            LOGGER.error(s"异常数据 : $line", e)
          }
        }
        logJson
      }).filter(f => {
      f != null && StringUtils.isNotEmpty(f.cid) && f.page.equals("/pages/content/detail/detail")
    })

    //处理
    val resTp2RDD = baseRDD.map(lj => {
      ((lj.cid, lj.pid), 1)
    }).reduceByKey(_ + _) // 先按照类别分组
      .groupBy(_._1._1).mapValues(itor => {
      var sortSet = new mutable.
                TreeSet[(((String, String), Int))]()(Ordering[Int].on(f => -f._2))
      itor.foreach(tp => {
        // 添加元素
        sortSet += tp
        if (sortSet.size > 2) sortSet remove (sortSet.last)
      })
      sortSet
    })

    resTp2RDD.foreach(println)

    sc.stop
  }
}
